from kafka import KafkaProducer
import json
import yaml
import random


file_path = "../../etc/config.yaml"
kafka_server = ''
kafka_topic = ''

with open(file_path, 'r', encoding='utf-8') as file:
    config = yaml.safe_load(file)
    kafka_config = config.get('kafka', {})
    kafka_server = kafka_config.get('server', '')
    kafka_topic = kafka_config.get('topic', '')

"""
/opt/kafka/bin/kafka-topics.sh --create --topic user_follow_topic --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1
"""

# 初始化 Kafka 生产者
producer = KafkaProducer(
    bootstrap_servers=kafka_server,
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)


def user_follow(follower_id, followed_id):
    message = {
        "action": random.choice(["user_follow", "user_unfollow"]),
        "follower_id": follower_id,
        "followed_id": followed_id
    }
    producer.send(kafka_topic, value=message)
    producer.flush()
    print(f"User {follower_id} followed user {followed_id}, message sent to Kafka.")


if __name__ == "__main__":
    [user_follow(i, j) for i in range(1, 5) for j in range(1, 5) if i != j]
